package cn.bby.query.server.infrastructure.impl.event;

import cn.bby.query.server.application.manager.DataChangeEventManager;
import cn.bby.query.server.application.manager.SystemStatusManager;
import cn.bby.query.server.domain.service.dto.DataChangeEvent;
import cn.bby.query.server.domain.share.constant.HealthInfoMetric;
import cn.bby.query.server.infrastructure.impl.event.model.Payload;
import cn.bby.query.server.infrastructure.impl.event.model.Source;
import cn.hutool.core.net.NetUtil;
import com.alibaba.fastjson2.JSON;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamAddArgs;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author wangbenyin
 * @version 1.0
 * @date 2025/06/05
 * @since 17
 */
@Slf4j
//@Configuration
public class RedisDataChangeEventManager extends DataChangeEventManager {

    public static final String GROUP_NAME = "adv-query-group";
    public static final String STREAM_NAME = "name-data-change";
    private final RStream<Integer, String> stream;
    private final List<Thread> threads = Collections.synchronizedList(new LinkedList<>());

    @Resource
    private SystemStatusManager systemStatusManager;
    @Value("${query.event.max-consumer-thread:2}")
    private int maxConsumerThread = 2;

    public static void main(String[] args) {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://localhost:10000")
                .setPassword("bjhne792sjoa");
        new RedisDataChangeEventManager(Redisson.create(config));
        while (true) {

        }
    }

    public RedisDataChangeEventManager(RedissonClient redissonClient) {
        stream = redissonClient.getStream(STREAM_NAME);
        stream.expire(10, TimeUnit.MINUTES);
        try {
            stream.createGroup(GROUP_NAME);
        } catch (Exception e) {
            log.warn("已有消费组");
        }
        for (int i = 1; i <= 2; i++) {
            Thread thread = new Thread(() -> runConsumer(stream));
            thread.setName("stream-consumer-" + i);
            thread.start();
            threads.add(thread);
        }
    }

    @Scheduled(fixedDelay = 20000, initialDelay = 60000)
    public void checkAndRun() {
        int sub = maxConsumerThread - threads.size();
        if (sub <= 0) {
            systemStatusManager.heartbeat(HealthInfoMetric.CONSUMER);
            return;
        }
        for (int i = sub; i <= maxConsumerThread; i++) {
            Thread thread = new Thread(() -> runConsumer(stream));
            thread.setName("stream-consumer-" + i);
            thread.start();
            threads.add(thread);
        }
    }

    private void runConsumer(RStream<Integer, String> stream) {
        String name = NetUtil.getLocalHostName() + "-" + Thread.currentThread().getName();
        while (!Thread.currentThread().isInterrupted()) {
            try {
                Map<StreamMessageId, Map<Integer, String>> messages =
                        stream.readGroup(GROUP_NAME, name,
                                StreamReadGroupArgs.neverDelivered()
                                        .count(10)
                                        .timeout(Duration.of(4, ChronoUnit.SECONDS)));
                if (messages == null || messages.isEmpty()) {
                    continue;
                }
                try {
                    List<DataChangeEvent> events = messages.values().stream()
                            .flatMap(map -> map.values().stream().map(payloadStr -> {
                        Payload payload = JSON.parseObject(payloadStr, Payload.class);
                        Source source = payload.source;
                        if (source == null || payload.op == null || payload.after == null && payload.before == null) {
                            return null;
                        }
                        DataChangeEvent event = new DataChangeEvent();
                        event.setTableFullName("%s.%s.%s".formatted(source.getDb(), source.getSchema(), source.getTable()));
                        event.setOpType(payload.op.toUpperCase(Locale.ROOT));
                        event.setBefore(payload.before);
                        event.setAfter(payload.after);
                        return event;
                    }).filter(e -> e != null && (e.getAfter() != null || e.getBefore() != null))).toList();
                    handleEvent(events);
                    stream.ackAsync(GROUP_NAME, messages.keySet().toArray(new StreamMessageId[0]));
                } catch (Exception e) {
                    log.error("处理消息异常", e);
                }
            } catch (Exception e) {
                log.error("读取消息异常", e);
                Thread.currentThread().interrupt();
                threads.remove(Thread.currentThread());
                return;
            }
        }
    }

    @Override
    public void sendEvent(List<String> payloads) {
        Map<Integer, String> map = new HashMap<>();
        for (int i = payloads.size() - 1; i >= 0; i--) {
            map.put(i, payloads.get(i));
        }
        stream.addAsync(StreamAddArgs.entries(map));
    }

    @PreDestroy
    public void clear() {
        for (Thread thread : threads) {
            thread.interrupt();
        }
        threads.clear();
    }
}
